package com.shujia.core

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector
import org.apache.flink.api.scala._
import org.apache.flink.api.common.time.Time

object Demo3StateWC {
  def main(args: Array[String]): Unit = {


    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val ds = env.socketTextStream("node1", 8888)

    val keyDS = ds
      .flatMap(_.split(","))
      .map(w => (w, 1))
      .keyBy(_._1)


    //keyDS.flatMap(new ReduceFlapMapFunaction).print()

    //scala特有api   值操作状态
    keyDS.mapWithState((kv, state: Option[Int]) => {
      val last = state.getOrElse(0)
      val curr = kv._2

      val sum = last + curr

      ((kv._1, sum), Some(sum))
    }).print()


    env.execute()

  }
}

class ReduceFlapMapFunaction extends RichFlatMapFunction[(String, Int), (String, Int)] {


  //保存当前key的状态
  // 状态最终会保存到checkpoint中
  var countState: ValueState[Int] = _

  //没一行数据都会执行一次
  override def flatMap(kv: (String, Int), out: Collector[(String, Int)]): Unit = {


    //获取状态中保存的值
    val last = countState.value()


    val curr = kv._2

    val sum = last + curr


    //更新状态中的值
    countState.update(sum)


    //返回结果
    out.collect((kv._1, sum))

  }

  //类被初始化的时候执行   最先执行
  override def open(parameters: Configuration): Unit = {
    //初始化状态
    val valueStateDescriptor = new ValueStateDescriptor[Int]("count", classOf[Int], 0)

    val ttlConfig = StateTtlConfig
      .newBuilder(Time.seconds(5))
      .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
      .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
      .build

    //启动TTL
    valueStateDescriptor.enableTimeToLive(ttlConfig)

    //获取状态对象
    countState = getRuntimeContext.getState(valueStateDescriptor)

  }
}
